package com.itcam.kafka.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * @author : Cammy.Wu
 * Description : 
 */

public class CustomProducerInterceptor implements ProducerInterceptor<String, String> {

    /**
     * 发送消息之前的切面拦截
     * @param producerRecord
     * @return
     */
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
        System.err.println("生产者发送前置方法！");
        String value = "prefix:" + producerRecord;
        ProducerRecord record = new ProducerRecord(
                producerRecord.topic(),
                producerRecord.partition(),
                producerRecord.timestamp(),
                producerRecord.key(),
                value,
                producerRecord.headers());
        return record;
    }

    /**
     * 发送消息之后的切面拦截
     * @param recordMetadata
     * @param e
     */
    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
        System.err.println("生产者发送后置方法！");
    }

    /**
     * 生产者关闭前调用的方法
     */
    @Override
    public void close() {
        System.err.println("close方法执行了！");
    }

    /**
     * 拦截器用于配置一些属性的方法
     * @param map
     */
    @Override
    public void configure(Map<String, ?> map) {

    }
}
